package com.cloud.rocketmqtxprovider.config.listener;

import com.alibaba.fastjson.JSONObject;
import com.cloud.rocketmqtxprovider.entity.Order;
import com.cloud.rocketmqtxprovider.service.ProviderService;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
@SuppressWarnings("all")
public class RocketMQProviderTransactionListener implements TransactionListener {

    @Autowired
    private ProviderService providerService;

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        //这里处理业务逻辑，业务逻辑和记录事务消息在同一个事务中进行，这里的事务控制是在 providerService 中的
        try{
            String orderMsg = new String(msg.getBody());
            Order order = JSONObject.parseObject(orderMsg, Order.class);
            providerService.insertOrder(order,msg.getTransactionId());
        }catch (Exception e){
            e.printStackTrace();
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        //这里是消息的轮询，当执行一些耗时的业务逻辑时，rocket会隔一段时间轮询一次，看看业务逻辑是否已经执行完
        if (providerService.selectTransactionCount(msg.getTransactionId()) > 0){
            //说明业务逻辑已经执行完了，可以提交了
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        return LocalTransactionState.UNKNOW;
    }
}
